package com.sc.byahi.user.server;

import static com.google.appengine.api.datastore.TransactionOptions.Builder.withXG;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collection;
import java.util.Collections;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.servlet.http.HttpSession;

import com.google.appengine.api.blobstore.BlobKey;
import com.google.appengine.api.blobstore.BlobstoreFailureException;
import com.google.appengine.api.blobstore.BlobstoreService;
import com.google.appengine.api.blobstore.BlobstoreServiceFactory;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.FetchOptions;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.Query;
import com.google.appengine.api.datastore.Query.CompositeFilterOperator;
import com.google.appengine.api.datastore.Query.FilterOperator;
import com.google.appengine.api.datastore.Text;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.api.images.ImagesService;
import com.google.appengine.api.images.ImagesServiceFactory;
import com.google.appengine.api.images.ServingUrlOptions;
import com.google.appengine.api.memcache.Expiration;
import com.google.appengine.api.memcache.MemcacheService;
import com.google.appengine.api.memcache.MemcacheServiceFactory;
import com.google.appengine.api.users.User;
import com.google.appengine.api.users.UserService;
import com.google.appengine.api.users.UserServiceFactory;
import com.google.apphosting.api.ApiProxy.CapabilityDisabledException;
import com.google.gwt.user.server.rpc.RemoteServiceServlet;
import com.sc.byahi.user.client.GreetingService;
import com.sc.byahi.user.shared.AbstractEntity;
import com.sc.byahi.user.shared.HasProperty;
import com.sc.byahi.user.shared.InitDataVO;
import com.sc.byahi.user.shared.Notice;
import com.sc.byahi.user.shared.Profile;
import com.sc.byahi.user.shared.ProfileMetadata;
import com.sc.byahi.user.shared.ProfileOwner;
import com.sc.byahi.user.shared.ProfileVO;
import com.sc.byahi.user.shared.SearchBean;
import com.sc.byahi.user.shared.SharedUtil;
import com.sc.byahi.user.shared.SuccessStory;
import com.sc.byahi.user.shared.constants.Config;
import com.sc.byahi.user.shared.constants.Lists;

/**
 * The server side implementation of the RPC service.
 */
@SuppressWarnings("serial")
public class GreetingServiceImpl extends RemoteServiceServlet implements GreetingService {
	private static Logger logger = Logger.getLogger("com.sc.byahi.user.server.GreetingServiceImpl");

	private static final String S_USER = "user";
	private static final String S_CURR_PROFILE_ID = "currentProfileId";

	@Deprecated
	private String getCurrentUser1() {
		HttpSession session = this.getThreadLocalRequest().getSession();
		String gmail = (String) session.getAttribute(S_USER);
		if (gmail == null) {
			gmail = this.getThreadLocalRequest().getUserPrincipal().getName();
			// UserServiceFactory.getUserService().getCurrentUser();
			session.setAttribute(S_USER, gmail);
		}
		return (String) session.getAttribute(S_USER);
	}

	private String getCurrentUser() {
		User user = UserServiceFactory.getUserService().getCurrentUser();
		String email = user.getEmail();
		return email.toLowerCase();
	}

	private Long getCurrentProfileId() {
		HttpSession session = this.getThreadLocalRequest().getSession();
		Long currentProfileId = (Long) session.getAttribute(S_CURR_PROFILE_ID);
		if (currentProfileId == null) {
			List<Long> myProfiles = findMyProfileIds(null);// any profile, approved or unapproved
			if (!myProfiles.isEmpty()) {
				session.setAttribute(S_CURR_PROFILE_ID, myProfiles.get(0));
				return myProfiles.get(0);
			}
		}
		return currentProfileId;
	}

	private void kickUserOut() {
		logger.log(Level.SEVERE, "Kicking out user: " + getCurrentUser());
		try {
			this.getThreadLocalResponse().sendRedirect(getLogoutURL());
			logger.log(Level.SEVERE, "Kick success.");
		} catch (IOException e) {
			logger.log(Level.SEVERE, "Kick failed, trying one more time.", e);
			try {
				this.getThreadLocalResponse().sendRedirect(getLogoutURL());
			} catch (IOException e1) {
				logger.log(Level.SEVERE, "Kick failed agian.", e);
			}
		}
	}

	private String getLogoutURL() {
		// HttpServletRequest request = this.getThreadLocalRequest();
		// String thisURL = request.getRequestURI();
		UserService userService = UserServiceFactory.getUserService();
		String url = "";
		if (StoreUtil.DEV_MODE) {
			// url = userService.createLogoutURL("/public/index.html?gwt.codesvr=127.0.0.1:9997");
			url = userService.createLogoutURL("/public/index.html");
		} else {
			url = userService.createLogoutURL("/public/index.html");
		}
		logger.log(Level.INFO, "Dev mode is: " + StoreUtil.DEV_MODE + " hence login URL is: " + url);
		return url;
	}

	@Override
	public String getUploadURL() {
		BlobstoreService blobstoreService = BlobstoreServiceFactory.getBlobstoreService();
		String url = new String();
		// url = blobstoreService.createUploadUrl("/user/upload");
		url = blobstoreService.createUploadUrl("/user/imageupload");
		url = SharedUtil.fixURL(url);
		return url;
	}

	@Override
	public String getPublicUploadURL() {
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			BlobstoreService blobstoreService = BlobstoreServiceFactory.getBlobstoreService();
			String url = new String();
			url = blobstoreService.createUploadUrl("/public/uploadbulk");
			return url;
		}
		return "";
	}

	private List<Key> createKeys(Collection<Long> ids, String className) {
		List<Key> keys = new ArrayList<Key>();
		for (Long long1 : ids) {
			Key k = KeyFactory.createKey(className, long1);
			keys.add(k);
		}
		return keys;
	}

	/**
	 * ------------------------------------------------------------------------------------------
	 * 
	 * ADMIN ONLY
	 * 
	 * ------------------------------------------------------------------------------------------
	 */

	/**
	 * Deletes key or keys, handles exceptions gracefully and return boolean. As, only Admin will delete data, so no worries.
	 * 
	 * @param datastore
	 *            can be NULL if there is no existing transaction.
	 * @param keys
	 * @return Boolean.FALSE if any one key failed.
	 * @deprecated
	 */
	private Boolean deleteKeys(DatastoreService datastore, Key... keys) {
		if (datastore == null)
			datastore = DatastoreServiceFactory.getDatastoreService();
		Boolean anyFailed = Boolean.FALSE;
		for (Key key : keys) {
			try {
				datastore.delete(key);
			} catch (ConcurrentModificationException e) {
				try {
					logger.log(Level.SEVERE, "Deleting key failed, trying one more time: " + key, e);
					datastore.delete(key);
				} catch (Exception e1) {
					logger.log(Level.SEVERE, "Deleting key failed: " + key, e1);
					anyFailed = Boolean.TRUE;
				}
			} catch (Exception e) {
				logger.log(Level.SEVERE, "Deleting key failed: " + key, e);
				anyFailed = Boolean.TRUE;
			}
		}
		return anyFailed;
	}

	/**
	 * Admin user will delete it
	 */
	@Override
	public Boolean deleteProfileOwner(String gmailId) {
		Boolean ret = Boolean.FALSE;
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			ProfileOwner owner = findProfileOwner(gmailId);
			if (owner != null) {
				logger.log(Level.SEVERE, "Trying to delete profile owner: " + gmailId);
				// All profiles
				List<Long> attachedprofiles = findProfileIdsForOwner(null, gmailId);
				for (Long id : attachedprofiles) {
					deleteProfileAndRelatedData(id);
				}
				Key ownerKey = KeyFactory.createKey(ProfileOwner.class.getSimpleName(), owner.getId());
				DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
				Transaction txn = datastore.beginTransaction(withXG(true));
				try {
					datastore.delete(ownerKey);
					ret = Boolean.TRUE;
				} finally {
					if (txn.isActive()) {
						txn.rollback();
					}
				}
			}
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to delete profile owner: " + gmailId);
			kickUserOut();
		}
		return ret;
	}

	/**
	 * Admin user only
	 */
	@Override
	public Boolean deleteProfile(Long id) {
		Boolean ret = Boolean.FALSE;
		if (UserServiceFactory.getUserService().isUserAdmin() || isMyProfileId(id)) {
			ret = deleteProfileAndRelatedData(id);
			clearMemcache();
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to delete profile : " + id);
			kickUserOut();
		}
		return ret;
	}

	/**
	 * Only admin user, or user himself, can delete profile Check for conditions before calling this method
	 * 
	 * @param id
	 * @param datastore
	 */
	@SuppressWarnings("unused")
	private Boolean deleteProfileAndRelatedData(Long id) {
		logger.log(Level.SEVERE, "Trying to delete profile : " + id);
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		// XG can operate on max 5 entity groups
		Transaction txn = datastore.beginTransaction(withXG(true));
		Boolean ret = Boolean.FALSE;
		try {
			List<Key> metadataKeys = new ArrayList<Key>();

			myMetadata: {
				Query q = new Query(ProfileMetadata.class.getSimpleName());
				q.setFilter(FilterOperator.EQUAL.of(ProfileMetadata.INIT_PROFILE_ID, id));
				// q.addFilter(ProfileMetadata.PROP_INIT_PROFILE, FilterOperator.EQUAL, id);
				q.setKeysOnly();
				Iterable<Entity> result = datastore.prepare(q).asIterable(FetchOptions.Builder.withDefaults());
				logger.log(Level.SEVERE, "Deleting my profile metadata");
				for (Entity entity : result) {
					datastore.delete(entity.getKey());
				}
				logger.log(Level.SEVERE, "Deleting my profile metadata - Success");
			}

			othersMetadata: {
				Query q = new Query(ProfileMetadata.class.getSimpleName());
				q.setFilter(FilterOperator.EQUAL.of(ProfileMetadata.TARGET_PROFILE_ID, id));
				// q.addFilter(ProfileMetadata.PROP_TARGET_PROFILE, FilterOperator.EQUAL, id);
				q.setKeysOnly();
				Iterable<Entity> result = datastore.prepare(q).asIterable(FetchOptions.Builder.withDefaults());
				logger.log(Level.SEVERE, "Deleting others profile metadata");
				for (Entity entity : result) {
					datastore.delete(entity.getKey());
				}
				logger.log(Level.SEVERE, "Deleting others profile metadata - Success");
			}

			searchbeans: {
				Query q = new Query(SearchBean.class.getSimpleName());
				q.setKeysOnly();
				q.setFilter(FilterOperator.EQUAL.of(SearchBean.OWNER, getCurrentUser()));
				// q.addFilter("owner", FilterOperator.EQUAL, getCurrentUser());

				logger.log(Level.SEVERE, "Deleting search bean");
				for (Entity e : datastore.prepare(q).asIterable()) {
					datastore.delete(e.getKey());
				}
				logger.log(Level.SEVERE, "Deleting search bean - Success");
			}

			// Now delete profile
			Key key = KeyFactory.createKey(Profile.class.getSimpleName(), id);

			photos: {
				try {
					Entity p = datastore.get(key);
					Object imagesObj = p.getProperty(Profile.IMAGES);
					if (imagesObj != null) {
						BlobstoreService blobstoreService = BlobstoreServiceFactory.getBlobstoreService();
						for (String imageKey : (List<String>) imagesObj) {
							BlobKey oldImageBlobKey = new BlobKey(imageKey);
							try {
								logger.log(Level.WARNING, "Deleting images: " + oldImageBlobKey.toString());
								blobstoreService.delete(oldImageBlobKey);
							} catch (BlobstoreFailureException bfe) {
								logger.log(Level.SEVERE, "Error while deleting Blobkey: " + oldImageBlobKey.toString(), bfe);
							}
						}
					}
				} catch (EntityNotFoundException e) {
					logger.log(Level.SEVERE, "Error while deleting profile: " + id, e);
				}
			}

			logger.log(Level.SEVERE, "Deleting profile");
			datastore.delete(key);
			logger.log(Level.SEVERE, "Deleting profilem - Success");
			ret = Boolean.TRUE;
		} finally {
			if (txn.isActive()) {
				txn.rollback();
			}
		}
		return ret;
	}

	/**
	 * Admin user only
	 */
	@Override
	public Boolean approveProfile(Long profileId) {
		// Only admin user has right to delete metadata or profile.
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			Boolean status = updateProfileApprovalStatus(profileId, true);
			clearMemcache();
			return status;
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to approve profile : " + profileId);
			kickUserOut();
		}
		return false;
	}

	/**
	 * Admin user or user for his own profile only
	 */
	@Override
	public Boolean disapproveProfile(Long profileId) {
		if (isMyProfileId(profileId) || UserServiceFactory.getUserService().isUserAdmin()) {
			Boolean status = updateProfileApprovalStatus(profileId, false);
			clearMemcache();
			return status;
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to disapprove profile : " + profileId);
			kickUserOut();
		}
		return false;
	}

	/**
	 * Admin user only, checks for amin user have already been done in calling methods. Note: PRIVATE method
	 */
	private Boolean updateProfileApprovalStatus(Long profileId, boolean status) {

		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Key key = KeyFactory.createKey(Profile.class.getSimpleName(), profileId);
		try {
			Entity e = datastore.get(key);
			e.setProperty(Profile.APPROVED, status);
			datastore.put(e);
		} catch (EntityNotFoundException e) {
			logger.log(Level.SEVERE, "Entity suddenly disappeared: " + key.getId(), e);
			return false;
		} catch (CapabilityDisabledException e) {
			return false;
		}

		return true;
	}

	@Override
	public List<Long> findAdminUnapprovedProfiles() {
		List<Long> result = new ArrayList<Long>();
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Query q = new Query(Profile.class.getSimpleName());
			q.setKeysOnly();
			q.setFilter(FilterOperator.EQUAL.of(Profile.APPROVED, false));

			List<Entity> results = datastore.prepare(q).asList(FetchOptions.Builder.withDefaults());
			for (Entity entity : results) {
				result.add(entity.getKey().getId());
			}
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to find unapproved profiles");
			kickUserOut();
		}
		return result;
	}

	@Override
	public String createTestProfiles(String gender, String data) {
		// Only admin user has right to create test profiles.
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			List<Profile> profiles = new TestProfileCreator().createTestProfiles(gender, data);
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			StringBuilder sb = new StringBuilder();
			for (Iterator<Profile> iterator = profiles.iterator(); iterator.hasNext();) {
				Profile profile = (Profile) iterator.next();

				ProfileOwner profileOwner = new ProfileOwner(profile.getOwner());
				profileOwner.setLastLogin(System.currentTimeMillis());
				profileOwner.setIsAgency(true);
				Entity p = new Entity(ProfileOwner.class.getSimpleName(), profileOwner.getId());
				setPropertiesToEntity(profileOwner, p);

				datastore.put(p);

				profile.setId(ProfileIdGenerator.get().getNewProfileId());
				Entity entity = new Entity(Profile.class.getSimpleName(), profile.getId());
				setPropertiesToEntity(profile, entity);

				datastore.put(entity);

				profile.setId(entity.getKey().getId());
				sb.append("\"").append(profile.getName()).append(",").append(profile.getId()).append("\"");
				if (iterator.hasNext()) {
					sb.append(",");
				}
			}
			return sb.toString();
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to create test profile");
			kickUserOut();
		}
		return "";
	}

	@Override
	public Boolean clearMemcache() {
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			MemcacheService syncCache = MemcacheServiceFactory.getMemcacheService();
			syncCache.clearAll();
			return Boolean.TRUE;
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to clear cache");
			kickUserOut();
		}
		return Boolean.FALSE;

	}

	@Override
	public String getMemcache() {
		Map<String, Object> data = new HashMap<String, Object>();
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			MemcacheService syncCache = MemcacheServiceFactory.getMemcacheService();
			// If memcache was cleared, then no object is there
			if (syncCache.contains("_KEYS_")) {
				@SuppressWarnings("unchecked")
				Set<String> keys = (Set<String>) syncCache.get("_KEYS_");
				for (String key : keys) {
					data.put(key, syncCache.get(key));
				}
			}
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to get memcache data");
			kickUserOut();
		}
		String base64 = com.sc.byahi.user.shared.Base64Utils.toBase64(data.toString().getBytes());
		return base64;
	}

	@Override
	public Boolean toggleMemcache() {
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			Config.memcache = !Config.memcache;
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to toggle memcache.");
			kickUserOut();
		}
		return Config.memcache;
	}

	@Override
	public boolean toggleTestUpload() {
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			Config.bulkTestUpload = !Config.bulkTestUpload;
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to toggle Test upload.");
			kickUserOut();
		}
		return Config.bulkTestUpload;
	}

	@Override
	public Boolean approveEmail(Long metadataId) {
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			try {
				DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
				Key k = KeyFactory.createKey(ProfileMetadata.class.getSimpleName(), metadataId);
				Entity e = datastore.get(k);
				Long pid = (Long) e.getProperty(ProfileMetadata.TARGET_PROFILE_ID);
				String body = (String) e.getProperty(ProfileMetadata.VALUE);

				Entity profileEntity = datastore.get(KeyFactory.createKey(Profile.class.getSimpleName(), pid));
				String email = (String) profileEntity.getProperty(Profile.OWNER);
				boolean sent = Mailer.sendMail(body, email);
				if (sent) {
					datastore.delete(k);
				}
				return true;
			} catch (EntityNotFoundException e1) {
				logger.log(Level.SEVERE, "Erorr while sending email: " + metadataId);
			}
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to approve email " + metadataId);
			kickUserOut();
		}
		return false;
	}

	@Override
	public Map<Long, String> getUnsentEmails() {
		Map<Long, String> mails = new HashMap<Long, String>();
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			List<Entity> md = findProfileMetadataEntities((Long) null, (Long) null, ProfileMetadata.ACTION_MAIL);
			for (Entity entity : md) {
				Long id = entity.getKey().getId();
				String body = (String) entity.getProperty(ProfileMetadata.VALUE);
				mails.put(id, body);
			}
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to get unsent emails");
			kickUserOut();
		}
		return mails;
	}

	@Override
	public void publishNotice(Notice notice) {
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			deleteNotice();
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Entity e = new Entity(Notice.class.getSimpleName());
			setPropertiesToEntity(notice, e);
			datastore.put(e);
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to publish notice");
			kickUserOut();
		}
	}

	@Override
	public void deleteNotice() {
		if (UserServiceFactory.getUserService().isUserAdmin()) {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Transaction txn = datastore.beginTransaction(withXG(true));
			Query q = new Query(Notice.class.getSimpleName());
			q.setKeysOnly();
			List<Entity> results = datastore.prepare(q).asList(FetchOptions.Builder.withDefaults());
			try {
				for (Entity entity : results) {
					datastore.delete(entity.getKey());
				}
				txn.commit();
			} finally {
				if (txn.isActive()) {
					txn.rollback();
				}
			}

		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to delete notice");
			kickUserOut();
		}
	}

	/**
	 * ------------------------------------------------------------------------------------------
	 * 
	 * ADMIN ONLY END
	 * 
	 * ------------------------------------------------------------------------------------------
	 */

	/**
	 * 
	 * @return
	 */
	private ProfileOwner getCurrProfileOwner() {
		String gmail = getCurrentUser();
		ProfileOwner profileOwner = findProfileOwner(gmail);
		if (profileOwner == null) {
			profileOwner = new ProfileOwner(gmail);
			profileOwner.setLastLogin(System.currentTimeMillis());
			profileOwner = createProfileOwner(profileOwner);
		}
		List<Long> profileIds = findMyProfileIds(Boolean.TRUE);
		profileOwner.setApprovedProfiles(profileIds);
		profileIds = findMyProfileIds(Boolean.FALSE);
		profileOwner.setUnapporvedProfiles(profileIds);
		return profileOwner;
	}

	/**
	 * Any changes should be implemented in create test profile data also.
	 * 
	 * @param profileOwner
	 * @return
	 */
	private ProfileOwner createProfileOwner(ProfileOwner profileOwner) {

		ProfileOwner owner = findProfileOwner(profileOwner.getId());
		if (owner != null) {// no existing record in database
			logger.log(Level.SEVERE, "Profile owner already exists: " + profileOwner.getId());
			return owner;
		} else {
			logger.log(Level.INFO, "New user logged in with mail id: " + profileOwner.getId());

			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Entity e = new Entity(ProfileOwner.class.getSimpleName(), profileOwner.getId());
			setPropertiesToEntity(profileOwner, e);

			datastore.put(e);
			profileOwner.setId(e.getKey().getName());
			return profileOwner;
		}
	}

	@Override
	public ProfileOwner updateProfileOwner(ProfileOwner owner) {
		ProfileOwner existingOwner = findProfileOwner(getCurrentUser());
		// Allow update only if existing owner is null.
		if (existingOwner.getIsAgency() == null)
			existingOwner.setProperty(ProfileOwner.IS_AGENCY, owner.getIsAgency());
		existingOwner.setProperty(ProfileOwner.LAST_LOGIN, owner.getLastLogin());

		try {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Entity e = datastore.get(KeyFactory.createKey(ProfileOwner.class.getSimpleName(), owner.getId()));
			setPropertiesToEntity(owner, e);
			datastore.put(e);
		} catch (EntityNotFoundException e1) {
			logger.log(Level.SEVERE, "User: " + getCurrentUser() + " tried to update profile with non existant ID: " + owner.getId());
		}
		return owner;
	}

	private ProfileOwner findProfileOwner(String gmailId) {
		try {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Key key = KeyFactory.createKey(ProfileOwner.class.getSimpleName(), gmailId);
			Entity e = datastore.get(key);
			ProfileOwner owner = new ProfileOwner(e.getKey().getName());
			owner.setProperties(e.getProperties());
			return owner;

		} catch (EntityNotFoundException e) {
			logger.log(Level.INFO, "User " + gmailId + " logged in for the first time.");
			return null;// first log in, no owner data
		}
	}

	private void setPropertiesToEntity(HasProperty propHolder, Entity entity) {
		if (StoreUtil.DEV_MODE) {
			// Because local datastore viewer does not show unindexed properties
			for (Entry<String, Object> prop : propHolder.getProperties().entrySet()) {
				entity.setProperty(prop.getKey(), prop.getValue());
			}
		} else {
			for (String prop : propHolder.getIndexedPropertyNames()) {
				Object value = propHolder.getProperty(prop);
				entity.setProperty(prop, value);
			}
			for (String prop : propHolder.getUnindexedPropertyNames()) {
				Object value = propHolder.getProperty(prop);
				entity.setUnindexedProperty(prop, value);
			}
		}
	}

	/**
	 * If any logic is changed here, change in create test profile method also.
	 */
	@Override
	public Long createProfile(Profile profile) {
		ProfileOwner owner = getCurrProfileOwner();
		profile.setOwner(owner.getId());
		profile.setOwnerAgency(owner.getIsAgency());
		profile.setApproved(false);// only after approval, profile changes will be visible
		profile.setId(ProfileIdGenerator.get().getNewProfileId());

		Entity entity = new Entity(Profile.class.getSimpleName(), profile.getId());
		setPropertiesToEntity(profile, entity);

		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		datastore.put(entity);
		Mailer.sendMail("Welcome to Byahi. <br/> Your profile has been created. <br/> Your profile ID is: " + profile.getId(), profile.getOwner());
		return entity.getKey().getId();
	}

	@Override
	public void updateProfile(Profile profile) {
		if (isMyProfileId(profile.getId()) || UserServiceFactory.getUserService().isUserAdmin()) {
			profile.setApproved(false);
			Key k = KeyFactory.createKey(Profile.class.getSimpleName(), profile.getId());
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			try {
				Entity e = datastore.get(k);

				setPropertiesToEntity(profile, e);
				datastore.put(e);
			} catch (EntityNotFoundException e1) {
				logger.log(Level.SEVERE, "User " + getCurrentUser() + " tried to update profile id: " + profile.getId());
			}
		} else {
			logger.log(Level.WARNING, "Unauthorized user " + getCurrentUser() + " tried to update profile" + profile.getId());
		}
	}

	/**
	 * This is the place where profile ID is coming from user.
	 */
	@Override
	public List<Long> verifyProfileIds(List<Long> ids) {
		List<Long> verified = new ArrayList<Long>();
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		String name = Profile.class.getSimpleName();
		for (Long id : ids) {
			try {
				@SuppressWarnings("unused")
				Entity e = datastore.get(KeyFactory.createKey(name, id));
				verified.add(id);
			} catch (EntityNotFoundException e) {
				logger.log(Level.SEVERE, "User " + getCurrentUser() + " send false id: " + id, e);
			}
		}
		return verified;
	}

	@Override
	public List<Profile> findPublicProfilesByIds(List<Long> ids) {
		if (ids.isEmpty())
			return Collections.emptyList();// because app engine doesn't like empty collections in query
		List<Key> keys = new ArrayList<Key>();
		String name = Profile.class.getSimpleName();
		for (Long id : ids) {
			keys.add(KeyFactory.createKey(name, id));
		}
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		List<Entity> results = new ArrayList<Entity>();
		List<List<Key>> parts = SharedUtil.divideList(keys);
		for (List<Key> list : parts) {
			Query q = new Query(Profile.class.getSimpleName());
			List<Query.Filter> l = new ArrayList<Query.Filter>();
			if (!UserServiceFactory.getUserService().isUserAdmin()) {
				l.add(FilterOperator.EQUAL.of(Profile.APPROVED, true));
			}
			l.add(FilterOperator.IN.of(Entity.KEY_RESERVED_PROPERTY, list));
			addFiltersToQuery(q, l);
			results.addAll(datastore.prepare(q).asList(FetchOptions.Builder.withDefaults()));
		}
		List<Profile> profileList = createProfilesFromEntities(results, true);
		return profileList;
	}

	@Override
	public List<ProfileVO> findPublicProfilesMapDetailsByIds(List<Long> ids) {
		if (ids.isEmpty())
			return Collections.emptyList();// because app engine doesn't like empty collections in query

		List<ProfileVO> profileVOList = new ArrayList<ProfileVO>();
		List<Key> keys = createKeys(ids, Profile.class.getSimpleName());

		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Map<Key, Entity> profileList = datastore.get(keys);
		for (Map.Entry<Key, Entity> e : profileList.entrySet()) {
			ProfileVO vo = new ProfileVO(e.getKey().getId(), null, (Double) e.getValue().getProperty(Profile.CURR_LATITUDE), (Double) e.getValue().getProperty(Profile.CURR_LONGITUDE));
			profileVOList.add(vo);
		}
		return profileVOList;
	}

	@Override
	public List<Long> findPublicProfileIds(SearchBean sb) {
		long startTime = System.currentTimeMillis();

		Set<Long> appResult = null;// because in the end condition is tested for null
		{
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.EQUAL.of(Profile.APPROVED, true));
			appResult = fetchFromMemcache(q);
		}
		Set<Long> genderResult = null;// because in the end condition is tested for null
		{
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.EQUAL.of(Profile.GENDER, sb.getGender()));
			genderResult = fetchFromMemcache(q);
		}
		Set<Long> phyStatusResult = null;// because in the end condition is tested for null
		{
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.EQUAL.of(Profile.PHYSICAL_STATUS, sb.getPhysicalStatus()));
			phyStatusResult = fetchFromMemcache(q);
		}
		Set<Long> heightResult = null;// because in the end condition is tested for null
		{
			if (sb.getHeightBegin() != null || sb.getHeightEnd() != null) {
				Query q = new Query(Profile.class.getSimpleName());
				List<Query.Filter> l = new ArrayList<Query.Filter>();
				// Height should never be null
				if (sb.getHeightBegin() != null) {
					l.add(FilterOperator.GREATER_THAN_OR_EQUAL.of(Profile.HEIGHT, sb.getHeightBegin()));
				}
				if (sb.getHeightEnd() != null) {
					l.add(FilterOperator.LESS_THAN_OR_EQUAL.of(Profile.HEIGHT, sb.getHeightEnd()));
				}
				addFiltersToQuery(q, l);

				heightResult = fetchFromMemcache(q);
			}

		}

		Set<Long> mtResult = null;// because in the end condition is tested for
		{
			Query q = new Query(Profile.class.getSimpleName());
			if (!"".equals(sb.getMotherTongue())) {
				q.setFilter(FilterOperator.EQUAL.of(Profile.MOTHER_TONGUE, sb.getMotherTongue()));
				mtResult = fetchFromMemcache(q);
			}
		}

		Set<Long> ageResult = null;// because in the end condition is tested for
		{
			if (sb.getAgeEnd() != null || sb.getAgeBegin() != null) {
				Query q = new Query(Profile.class.getSimpleName());
				List<Query.Filter> l = new ArrayList<Query.Filter>();
				if (sb.getAgeEnd() != null) {
					Calendar last = getDate(sb.getAgeEnd().intValue());
					l.add(FilterOperator.GREATER_THAN_OR_EQUAL.of(Profile.DOB, last.getTime()));
				}
				if (sb.getAgeBegin() != null) {
					Calendar first = getDate(sb.getAgeBegin().intValue());
					l.add(FilterOperator.LESS_THAN_OR_EQUAL.of(Profile.DOB, first.getTime()));
				}
				addFiltersToQuery(q, l);
				ageResult = fetchFromMemcache(q);
			}
		}

		Set<Long> msResult = null;// because in the end condition is tested for
		if (!sb.getMaritalStatus().isEmpty()) {
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.IN.of(Profile.MARITAL_STATUS, sb.getMaritalStatus()));
			msResult = fetchFromMemcache(q);
		}

		Set<Long> religionResult = null;// because in the end condition is tested for null
		{
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.EQUAL.of(Profile.RELIGION, sb.getReligion()));
			religionResult = fetchFromMemcache(q);
		}

		Set<Long> casteResult = null;// because in the end condition is tested for
		if (!sb.getCaste().isEmpty()) {
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.IN.of(Profile.CASTE, sb.getCaste()));
			casteResult = fetchFromMemcache(q);
		}

		Set<Long> langResult = null;// because in the end condition is tested for
		if (!sb.getLanguages().isEmpty()) {
			Query q = new Query(Profile.class.getSimpleName());

			List<Query.Filter> l = new ArrayList<Query.Filter>();
			for (String language : sb.getLanguages()) {
				l.add(FilterOperator.EQUAL.of(Profile.MOTHER_TONGUE, language));
			}
			if (l.size() == 0) {
				q.setFilter(FilterOperator.IN.of(Profile.LANGUAGES, sb.getLanguages()));
			}
			if (l.size() == 1) {
				q.setFilter(CompositeFilterOperator.OR.of(FilterOperator.IN.of(Profile.LANGUAGES, sb.getLanguages()), l.get(0)));
			} else {
				q.setFilter(CompositeFilterOperator.OR.of(FilterOperator.IN.of(Profile.LANGUAGES, sb.getLanguages()), CompositeFilterOperator.OR.of(l)));
			}

			langResult = fetchFromMemcache(q);
		}

		Set<Long> degreeResult = null;// because in the end condition is tested for
		if (!sb.getDegrees().isEmpty()) {
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.IN.of(Profile.DEGREES, sb.getDegrees()));
			degreeResult = fetchFromMemcache(q);
		}

		Set<Long> dietResult = null;// because in the end condition is tested for
		if (!sb.getEatingHabits().isEmpty()) {
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.IN.of(Profile.EATING_HABITS, sb.getEatingHabits()));
			dietResult = fetchFromMemcache(q);
		}

		Set<Long> countryResult = null;// because in the end condition is tested for
		if (!sb.getLocationCountry().isEmpty()) {
			Query q = new Query(Profile.class.getSimpleName());
			Properties prop = ConstantsManager.getInstance().getCountryCodeList();
			List<String> countryCodes = new ArrayList<String>();
			for (String countryName : sb.getLocationCountry()) {
				countryCodes.add(prop.getProperty(countryName));
			}
			q.setFilter(FilterOperator.IN.of(Profile.CURR_COUNTRY, countryCodes));
			countryResult = fetchFromMemcache(q);
		}

		Set<Long> jobResult = null;// because in the end condition is tested for
		if (!sb.getEmploymentType().isEmpty()) {
			Query q = new Query(Profile.class.getSimpleName());
			q.setFilter(FilterOperator.IN.of(Profile.EMPLOYMENT_TYPE, sb.getEmploymentType()));
			jobResult = fetchFromMemcache(q);
		}

		long collectionStartTime = System.currentTimeMillis();

		Set<Long> s = new HashSet<Long>(appResult);

		if (genderResult != null)
			s.retainAll(genderResult);
		if (phyStatusResult != null)
			s.retainAll(phyStatusResult);
		if (heightResult != null)
			s.retainAll(heightResult);
		if (mtResult != null)
			s.retainAll(mtResult);
		if (ageResult != null)
			s.retainAll(ageResult);
		if (msResult != null)
			s.retainAll(msResult);
		if (religionResult != null)
			s.retainAll(religionResult);
		if (casteResult != null)
			s.retainAll(casteResult);
		if (langResult != null)
			s.retainAll(langResult);
		if (degreeResult != null)
			s.retainAll(degreeResult);
		if (dietResult != null)
			s.retainAll(dietResult);
		if (countryResult != null)
			s.retainAll(countryResult);
		if (jobResult != null)
			s.retainAll(jobResult);

		long endTime = System.currentTimeMillis();
		logger.log(Level.INFO, "Total search time: " + (endTime - startTime));
		logger.log(Level.INFO, "Totale collection time: " + (endTime - collectionStartTime));

		List<Long> ids = new ArrayList<Long>((Collection<Long>) s);

		if (ids.size() > Config.maxResultSize) {
			return ids.subList(0, Config.maxResultSize);
		}
		return ids;
	}

	/**
	 * All profiles where rating is less than specified should be removed, at the same time, unrated profiles should be preserved. So this method returns all profiles having rating below specified
	 * rating
	 * 
	 * @param upToRating
	 * @param myProfileId
	 * @return
	 * @deprecated
	 */
	private Set<Long> fetchRatingsFromDB(List<Long> ratings, Long myProfileId) {
		Set<Long> ids = new HashSet<Long>();
		if (!ratings.isEmpty()) {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Query q = new Query(ProfileMetadata.class.getSimpleName());

			List<Query.Filter> l = new ArrayList<Query.Filter>();
			l.add(FilterOperator.EQUAL.of(ProfileMetadata.INIT_PROFILE_ID, myProfileId));
			l.add(FilterOperator.EQUAL.of(ProfileMetadata.ACTION, ProfileMetadata.ACTION_RATE));
			l.add(FilterOperator.IN.of(ProfileMetadata.VALUE, ratings));
			q.setFilter(CompositeFilterOperator.AND.of(l));

			Set<Long> ratingResult = new HashSet<Long>();
			for (Entity entity : datastore.prepare(q).asIterable()) {
				Long taregtProfileId = (Long) entity.getProperty(ProfileMetadata.TARGET_PROFILE_ID);
				ratingResult.add(taregtProfileId);
			}
			ids.addAll(ratingResult);
		}
		return ids;
	}

	@Override
	public Boolean rateProfile(Long targetProfileId, Long rating) {
		Long myProfileId = getCurrentProfileId();
		List<Entity> md = findProfileMetadataEntities(myProfileId, targetProfileId, ProfileMetadata.ACTION_RATE);
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Transaction txn = datastore.beginTransaction(withXG(true));
		Boolean ret = Boolean.FALSE;
		try {
			Entity e = null;
			if (!md.isEmpty()) {
				if (md.size() > 1) {
					// do some housekeeping
					e = md.remove(0);// keep first
					for (Entity entity : md) {
						datastore.delete(entity.getKey());
					}
				}
				e = md.get(0);
			} else {
				e = new Entity(ProfileMetadata.class.getSimpleName());
			}
			e.setProperty(ProfileMetadata.INIT_PROFILE_ID, myProfileId);
			e.setProperty(ProfileMetadata.TARGET_PROFILE_ID, targetProfileId);
			e.setProperty(ProfileMetadata.ACTION, ProfileMetadata.ACTION_RATE);
			e.setProperty(ProfileMetadata.VALUE, rating);
			datastore.put(e);
			txn.commit();
			ret = Boolean.TRUE;
		} finally {
			if (txn.isActive()) {
				txn.rollback();
			}
		}
		return ret;

	}

	private void addFiltersToQuery(Query q, List<Query.Filter> l) {
		if (l.size() == 1) {
			q.setFilter(l.get(0));
		} else if (l.size() > 1) {
			q.setFilter(CompositeFilterOperator.AND.of(l));
		} else {
			// do nothing
		}
	}

	private Calendar getDate(int reduceYears) {
		Calendar cal = Calendar.getInstance();
		int year = cal.get(Calendar.YEAR);
		int month = cal.get(Calendar.MONTH);
		int day = cal.get(Calendar.DATE);
		cal.clear();
		cal.set(Calendar.YEAR, year - reduceYears);
		cal.set(Calendar.MONDAY, month);
		cal.set(Calendar.DAY_OF_MONTH, day);
		return cal;
	}

	@SuppressWarnings("unchecked")
	private Set<Long> fetchFromMemcache(Query q) {
		// This ensures that only keys sre stored in memcache
		q.setKeysOnly();
		Set<Long> idSet = new HashSet<Long>();
		if (Config.memcache) {
			// Using the synchronous cache, stores only Keys :)
			MemcacheService syncCache = MemcacheServiceFactory.getMemcacheService();
			String query = q.toString();

			if (syncCache.contains(query)) {
				logger.log(Level.INFO, "REUSED: " + query);
				idSet = (Set<Long>) syncCache.get(query); // read from cache
			} else {
				logger.log(Level.INFO, "NEW: " + query);
				idSet = executeQuery(q);
				putInMemcache(syncCache, query, idSet, Expiration.byDeltaSeconds(Config.memcacheExp)); // populate cache
			}
		} else {
			idSet = executeQuery(q);
		}

		return idSet;
	}

	/**
	 * Puts the key value in memcache and updates the keys part from memcache so that it could be used for disgnostic purposes
	 */

	@SuppressWarnings("unchecked")
	private void putInMemcache(MemcacheService syncCache, String key, Object value, Expiration exp) {
		if (exp == null) {
			syncCache.put(key, value);
		} else {
			syncCache.put(key, value, exp);
		}
		Set<String> keys = new HashSet<String>();
		if (syncCache.contains("_KEYS_")) {
			keys = (Set<String>) syncCache.get("_KEYS_");
		}
		keys.add(key);
		syncCache.put("_KEYS_", keys);
	}

	private Set<Long> executeQuery(Query q) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		List<Entity> result = datastore.prepare(q).asList(FetchOptions.Builder.withLimit(Config.resultCount));
		Set<Long> idSet = new HashSet<Long>();
		for (Entity entity : result) {
			idSet.add(entity.getKey().getId());
		}
		return idSet;
	}

	/**
	 * Upload servlet needs to get owner when it updates profile. If owner is cleared then the updated record will not have any owner. Hence clearOwner field is provided for Upload Servlet. method is
	 * friendly.
	 * 
	 * @param ids
	 * @return
	 */
	List<Profile> findProfiles(List<Long> ids, boolean clearOwner) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();

		List<Key> keys = createKeys(ids, Profile.class.getSimpleName());

		Map<Key, Entity> map = datastore.get(keys);

		return createProfilesFromEntities(new ArrayList<Entity>(map.values()), clearOwner);
	}

	@Override
	public List<Long> findPublicProfileIdsByAgency(Long sampleId) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Key key = KeyFactory.createKey(Profile.class.getSimpleName(), sampleId);
		try {
			Entity p = datastore.get(key);
			String owner = (String) p.getProperty(Profile.OWNER);

			Query q = new Query(Profile.class.getSimpleName());
			q.setKeysOnly();

			List<Query.Filter> l = new ArrayList<Query.Filter>();
			l.add(FilterOperator.EQUAL.of(Profile.OWNER, owner));
			l.add(FilterOperator.EQUAL.of(Profile.APPROVED, true));
			addFiltersToQuery(q, l);

			// Low usage query, no memcache needed
			List<Entity> results = datastore.prepare(q).asList(FetchOptions.Builder.withDefaults());
			List<Long> profileList = new ArrayList<Long>();
			for (Entity entity : results) {
				profileList.add(entity.getKey().getId());
			}
			return profileList;

		} catch (EntityNotFoundException e) {
			logger.log(Level.SEVERE, "User: " + getCurrentUser() + " tried to find profile with id: " + sampleId, e);
		}
		return new ArrayList<Long>();
	}

	/**
	 * 
	 * @param approved
	 *            NULL if you want to search all profiles for this owner
	 * @return
	 */
	@Override
	public List<Long> findMyProfileIds(Boolean approved) {
		String owner = getCurrentUser();
		if (owner != null) {
			return findProfileIdsForOwner(approved, owner);
		}
		return new ArrayList<Long>();
	}

	private List<Long> findProfileIdsForOwner(Boolean approved, String owner) {
		List<Long> profileIds = new ArrayList<Long>();
		Query q = new Query(Profile.class.getSimpleName());
		q.setKeysOnly();

		List<Query.Filter> l = new ArrayList<Query.Filter>();
		l.add(FilterOperator.EQUAL.of(Profile.OWNER, owner));
		if (approved != null) {
			l.add(FilterOperator.EQUAL.of(Profile.APPROVED, approved));
		}
		addFiltersToQuery(q, l);
		// Not memcache as low use query
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Iterable<Entity> results = datastore.prepare(q).asIterable(FetchOptions.Builder.withDefaults());
		for (Entity entity : results) {
			profileIds.add(entity.getKey().getId());
		}
		return profileIds;
	}

	@Override
	public List<Profile> findMyProfilesById(List<Long> ids) {
		List<Long> allProfileIds = findMyProfileIds(null);// all profiles, approved as well as unapproved

		Set<Long> s = new HashSet<Long>();
		s.addAll(ids);
		s.retainAll(allProfileIds);
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();

		List<Key> keys = createKeys(s, Profile.class.getSimpleName());

		try {
			Map<Key, Entity> entities = datastore.get(keys);

			List<Profile> profiles = createProfilesFromEntities(entities.values(), false);
			return profiles;
		} catch (Exception e) {
			logger.log(Level.SEVERE, "My profile with ids: " + ids + " threw Exception.", e);
		}
		return new ArrayList<Profile>();
	}

	/**
	 * This method is used for fetching profile names for all profiles, to be used in sumary view.
	 */
	@Override
	public Map<Long, String> findMyProfilesNamesByIds() {

		String owner = getCurrentUser();
		Map<Long, String> profileSummary = new HashMap<Long, String>();
		if (owner != null) {
			Query q = new Query(Profile.class.getSimpleName());
			List<Query.Filter> l = new ArrayList<Query.Filter>();
			l.add(FilterOperator.EQUAL.of(Profile.OWNER, owner));
			addFiltersToQuery(q, l);
			// Not memcache as low use query
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Iterable<Entity> results = datastore.prepare(q).asIterable(FetchOptions.Builder.withDefaults());
			for (Entity entity : results) {
				profileSummary.put(entity.getKey().getId(), (String) entity.getProperty(Profile.NAME));
			}
		}
		return profileSummary;
	}

	/**
	 * Friendly access so that upload servlet can check the validity.
	 * 
	 * @param id
	 * @return
	 */
	boolean isMyProfileId(Long id) {
		Key key = KeyFactory.createKey(Profile.class.getSimpleName(), id);
		String gmail = getCurrentUser();
		try {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Entity e = datastore.get(key);
			String owner = (String) e.getProperty(Profile.OWNER);
			if (gmail.equals(owner)) {
				return true;
			}
		} catch (EntityNotFoundException e) {
			logger.log(Level.SEVERE, "Profile with ID: " + id + " not found.", e);
		}
		return false;
	}

	private List<Profile> createProfilesFromEntities(Collection<Entity> results, boolean deleteOwner) {
		List<Profile> profileList = new ArrayList<Profile>();
		if (!results.isEmpty()) {

			for (Entity result : results) {
				Profile p = new Profile();
				Object aboutMe = result.getProperty(Profile.ABOUT_ME);
				String aboutMeStr = "";
				if (aboutMe instanceof String) {
					aboutMeStr = (String) aboutMe;
				} else if (aboutMe instanceof Text) {
					aboutMeStr = ((Text) aboutMe).getValue();
				}
				result.removeProperty(Profile.ABOUT_ME);
				p.setProperties(result.getProperties());
				p.setId(result.getKey().getId());
				p.setAboutMe(aboutMeStr);

				if (deleteOwner) {
					p.setOwner("");
				}
				if (Config.imageFromURL) {
					List<String> keys = p.getImages();
					List<String> urls = new ArrayList<String>();
					ImagesService imagesService = ImagesServiceFactory.getImagesService();
					for (String key : keys) {
						String URL = imagesService.getServingUrl(ServingUrlOptions.Builder.withBlobKey(new BlobKey(key)));
						// String URL = imagesService.getServingUrl(new BlobKey(key));
						URL = SharedUtil.fixURL(URL);
						urls.add(URL);
						logger.log(Level.INFO, "Key " + key + " is converted to URL: " + URL);
					}
					p.setImageURLs(urls);
				}

				profileList.add(p);
			}
		}
		return profileList;
	}

	@Override
	public Boolean interestShow(Long myProfileId, Long targetProfileId) {
		if (isMyProfileId(myProfileId)) {
			// check whether target has also sent an interest, if yest, accept it.
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			List<Entity> result = findProfileMetadataEntities(targetProfileId, myProfileId, ProfileMetadata.ACTION_INTEREST);
			Entity e = null;
			if (!result.isEmpty()) {
				e = result.get(0);
				logger.log(Level.WARNING, "Finally!!! one action where concurrency showed up.");
				e.setProperty(ProfileMetadata.VALUE, ProfileMetadata.INTEREST_VALUE_ACCEPTED);
				datastore.put(e);
			} else {
				e = new Entity(ProfileMetadata.class.getSimpleName());
				e.setProperty(ProfileMetadata.INIT_PROFILE_ID, myProfileId);
				e.setProperty(ProfileMetadata.TARGET_PROFILE_ID, targetProfileId);
				e.setProperty(ProfileMetadata.ACTION, ProfileMetadata.ACTION_INTEREST);
				e.setProperty(ProfileMetadata.VALUE, ProfileMetadata.INTEREST_VALUE_SENT);
				datastore.put(e);
			}
			return true;
		}
		return false;
	}

	@Override
	public Boolean interestAccept(Long myProfileId, Long senderProfileId) {
		if (isMyProfileId(myProfileId)) {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			List<Entity> result = findProfileMetadataEntities(senderProfileId, myProfileId, ProfileMetadata.ACTION_INTEREST);
			Entity e = result.get(0);
			e.setProperty(ProfileMetadata.VALUE, ProfileMetadata.INTEREST_VALUE_ACCEPTED);
			datastore.put(e);
			return true;

		}

		return false;
	}

	@Override
	public Boolean interestHold(Long myProfileId, Long senderProfileId) {
		if (isMyProfileId(myProfileId)) {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			List<Entity> result = findProfileMetadataEntities(senderProfileId, myProfileId, ProfileMetadata.ACTION_INTEREST);
			Entity e = result.get(0);
			e.setProperty(ProfileMetadata.VALUE, ProfileMetadata.INTEREST_VALUE_ON_HOLD);
			datastore.put(e);
			return true;
		}
		return false;
	}

	private List<Entity> findProfileMetadataEntities(Long sourceId, Long targetId, String action) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Query q = new Query(ProfileMetadata.class.getSimpleName());
		List<Query.Filter> l = new ArrayList<Query.Filter>();
		if (sourceId != null)
			l.add(FilterOperator.EQUAL.of(ProfileMetadata.INIT_PROFILE_ID, sourceId));
		if (targetId != null)
			l.add(FilterOperator.EQUAL.of(ProfileMetadata.TARGET_PROFILE_ID, targetId));
		l.add(FilterOperator.EQUAL.of(ProfileMetadata.ACTION, action));
		addFiltersToQuery(q, l);

		return datastore.prepare(q).asList(FetchOptions.Builder.withDefaults());
	}

	/**
	 * Divides the list into batches and then collects the result. APp engine allows only 30 suery paramateres in IN parameter
	 * 
	 * @param sourceId
	 * @param targetId
	 * @param action
	 * @return
	 */
	private List<Entity> findProfileMetadataEntities(List<Long> sourceId, List<Long> targetId, String action) {
		if (sourceId != null && !sourceId.isEmpty() && targetId != null && !targetId.isEmpty()) {
			Collections.emptyList();// Both MUST NOT be available
		}
		List<Entity> entities = new ArrayList<Entity>();
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		if (sourceId != null && !sourceId.isEmpty()) {
			List<List<Long>> parts = SharedUtil.divideList(sourceId);
			for (List<Long> list : parts) {
				Query q = new Query(ProfileMetadata.class.getSimpleName());
				List<Query.Filter> l = new ArrayList<Query.Filter>();
				l.add(FilterOperator.IN.of(ProfileMetadata.INIT_PROFILE_ID, list));
				l.add(FilterOperator.EQUAL.of(ProfileMetadata.ACTION, action));
				addFiltersToQuery(q, l);
				entities.addAll(datastore.prepare(q).asList(FetchOptions.Builder.withDefaults()));
			}
		} else if (targetId != null && !targetId.isEmpty()) {
			List<List<Long>> parts = SharedUtil.divideList(targetId);
			for (List<Long> list : parts) {
				Query q = new Query(ProfileMetadata.class.getSimpleName());
				List<Query.Filter> l = new ArrayList<Query.Filter>();
				l.add(FilterOperator.IN.of(ProfileMetadata.TARGET_PROFILE_ID, list));
				l.add(FilterOperator.EQUAL.of(ProfileMetadata.ACTION, action));
				addFiltersToQuery(q, l);
				entities.addAll(datastore.prepare(q).asList(FetchOptions.Builder.withDefaults()));
			}
		}
		return entities;
	}

	@Override
	public InitDataVO getInitData() {
		InitDataVO data = new InitDataVO();
		data.setCurrentUser(getCurrentUser());
		UserService userService = UserServiceFactory.getUserService();
		data.setCurrentUserAdmin(userService.isUserAdmin());
		data.setLogoutURL(getLogoutURL());
		data.setOwner(getCurrProfileOwner());

		data.setSbList(findSearchCriteriaNames());
		// data.setMetadata(findMyMetadata());
		data.setInterestData(findMyInterestData());

		data.setProfileNames(findMyProfilesNamesByIds());
		return data;
	}

	@Override
	public Map<Long, Long> setCurrentProfileId(Long currentProfileId) {
		if (currentProfileId != null && isMyProfileId(currentProfileId)) {
			HttpSession session = this.getThreadLocalRequest().getSession();
			session.setAttribute(S_CURR_PROFILE_ID, currentProfileId);
		}
		return findMyRatings();
	}

	/**
	 * Only for individual users Map<Long, Long> taregtProfileId rating
	 * 
	 * @return
	 */
	private Map<Long, Long> findMyRatings() {
		Map<Long, Long> ratings = new HashMap<Long, Long>();
		Long myProfileId = getCurrentProfileId();
		// First time logged in user will have no current profile ID
		if (myProfileId != null) {
			List<Entity> result = findProfileMetadataEntities(myProfileId, null, ProfileMetadata.ACTION_RATE);
			for (Entity entity : result) {
				Long taregtProfileId = (Long) entity.getProperty(ProfileMetadata.TARGET_PROFILE_ID);
				Long rating = (Long) entity.getProperty(ProfileMetadata.VALUE);
				ratings.put(taregtProfileId, rating);
			}
		}
		return ratings;
	}

	/**
	 * Finds interest data for user profiles. Generally unapproved profiles don't have interest data, but sometimes, when user modifies a profile, then the interest data will still exist.
	 * 
	 * @param profiles
	 * @return
	 */
	@Override
	public Map<Long, Map<Long, String>> findMyInterestData() {
		// List<Long> profiles = findMyProfileIds(Boolean.TRUE);// only for approved profiles.
		Map<Long, Map<Long, String>> allProfileInterestMap = new HashMap<Long, Map<Long, String>>();
		List<Long> profiles = findMyProfileIds(null);// all profiles
		if (profiles.isEmpty())
			return allProfileInterestMap;// empty map
		for (Long currProfileId : profiles) {
			Map<Long, String> interestMap = new HashMap<Long, String>();
			allProfileInterestMap.put(currProfileId, interestMap);
		}

		// Where ever this profile is source
		isSource: {
			List<Entity> mds = findProfileMetadataEntities(profiles, null, ProfileMetadata.ACTION_INTEREST);
			for (Entity md : mds) {
				Long myProfileId = (Long) md.getProperty(ProfileMetadata.INIT_PROFILE_ID);
				Long targetProfileId = (Long) md.getProperty(ProfileMetadata.TARGET_PROFILE_ID);
				Map<Long, String> interestMap = allProfileInterestMap.get(myProfileId);
				interestMap.put(targetProfileId, translateAction(true, (String) md.getProperty(ProfileMetadata.VALUE)));
			}
		}

		// Where ever this profile is target
		isTaregt: {
			List<Entity> mds = findProfileMetadataEntities(null, profiles, ProfileMetadata.ACTION_INTEREST);
			for (Entity md : mds) {
				Long initProfileId = (Long) md.getProperty(ProfileMetadata.INIT_PROFILE_ID);
				Long myProfileId = (Long) md.getProperty(ProfileMetadata.TARGET_PROFILE_ID);
				Map<Long, String> interestMap = allProfileInterestMap.get(myProfileId);
				interestMap.put(initProfileId, translateAction(false, (String) md.getProperty(ProfileMetadata.VALUE)));
			}
		}
		return allProfileInterestMap;
	}

	private String translateAction(boolean isSource, String action) {
		if (action.equals(ProfileMetadata.INTEREST_VALUE_SENT) && isSource) {
			return ProfileMetadata.IS;
		}
		if (action.equals(ProfileMetadata.INTEREST_VALUE_ACCEPTED) && isSource) {
			return ProfileMetadata.IA;
		}
		if (action.equals(ProfileMetadata.INTEREST_VALUE_ON_HOLD) && isSource) {
			return ProfileMetadata.IH;
		}
		if (action.equals(ProfileMetadata.INTEREST_VALUE_SENT) && !isSource) {
			return ProfileMetadata.RR;
		}
		if (action.equals(ProfileMetadata.INTEREST_VALUE_ACCEPTED) && !isSource) {
			return ProfileMetadata.RA;
		}
		if (action.equals(ProfileMetadata.INTEREST_VALUE_ON_HOLD) && !isSource) {
			return ProfileMetadata.RH;
		}
		return null;
	}

	@Override
	public Map<Lists, Map<String, List<String>>> getLists() {
		return ConstantsManager.getInstance().getLists();
	}

	// -------------------- SAVED CRITERIA BEGIN --------------------

	@Override
	public Boolean saveSearchCriteria(SearchBean sb) {
		List<Entity> results = findSearchBeanEntities(getCurrentUser(), sb.getName());
		Boolean ret = Boolean.FALSE;
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Transaction txn = datastore.beginTransaction(withXG(true));
		try {
			for (Entity entity : results) {
				datastore.delete(entity.getKey());
			}
			sb.setOwner(getCurrentUser());
			Entity e = new Entity(SearchBean.class.getSimpleName());
			setPropertiesToEntity(sb, e);
			datastore.put(e);
			txn.commit();
			ret = Boolean.TRUE;
		} finally {
			if (txn.isActive()) {
				txn.rollback();
			}
		}
		return ret;
	}

	@Override
	public List<String> findSearchCriteriaNames() {
		List<Entity> results = findSearchBeanEntities(getCurrentUser(), null);
		List<String> names = new ArrayList<String>();
		for (Entity e : results) {
			String name = (String) e.getProperty(SearchBean.NAME);
			names.add(name);
		}
		return names;
	}

	@Override
	public SearchBean findSearchCriteria(String name) {
		List<Entity> results = findSearchBeanEntities(getCurrentUser(), name);
		List<SearchBean> sbList = createSearchBeanFromEntities(results);
		if (sbList.size() > 0)
			return sbList.get(0);
		else
			return null;
	}

	@Override
	public void deleteSearchCriteria(String name) {
		List<Entity> results = findSearchBeanEntities(getCurrentUser(), name);
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Transaction txn = datastore.beginTransaction(withXG(true));
		try {
			for (Entity entity : results) {
				datastore.delete(entity.getKey());
			}
			txn.commit();
		} finally {
			if (txn.isActive()) {
				txn.rollback();
			}
		}
	}

	private List<Entity> findSearchBeanEntities(String owner, String name) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Query q = new Query(SearchBean.class.getSimpleName());
		List<Query.Filter> l = new ArrayList<Query.Filter>();
		if (owner != null)
			l.add(FilterOperator.EQUAL.of(SearchBean.OWNER, owner));
		if (name != null)
			l.add(FilterOperator.EQUAL.of(SearchBean.NAME, name));
		addFiltersToQuery(q, l);
		return datastore.prepare(q).asList(FetchOptions.Builder.withDefaults());
	}

	private List<SearchBean> createSearchBeanFromEntities(List<Entity> results) {
		List<SearchBean> sbList = new ArrayList<SearchBean>();
		for (Entity result : results) {
			SearchBean sb = new SearchBean();
			sb.setId(result.getKey().getId());
			sb.setProperties(result.getProperties());
			sbList.add(sb);
		}
		return sbList;
	}

	// -------------------- SAVED CRITERIA END --------------------
	@Override
	public Boolean sendMessage(Long from, Long to, String body) {
		if (body.length() > 300) {
			body = body.substring(0, 299);
		}

		// TODO: whether user can send the message to this profile or not.
		int acc1 = findProfileMetadataEntities(from, to, ProfileMetadata.ACTION_INTEREST).size();
		int acc2 = findProfileMetadataEntities(to, from, ProfileMetadata.ACTION_INTEREST).size();
		if (acc1 > 0 || acc2 > 0) {
			DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
			Transaction txn = datastore.beginTransaction(withXG(true));
			try {
				List<Entity> md = findProfileMetadataEntities(from, to, ProfileMetadata.ACTION_MAIL);
				Entity e = null;
				if (!md.isEmpty()) {
					if (md.size() > 1) {
						// do some housekeeping
						e = md.remove(0);// keep first
						for (Entity entity : md) {
							datastore.delete(entity.getKey());
						}
					}
					e = md.get(0);
				} else {
					e = new Entity(ProfileMetadata.class.getSimpleName());
				}
				e.setProperty(ProfileMetadata.INIT_PROFILE_ID, from);
				e.setProperty(ProfileMetadata.TARGET_PROFILE_ID, to);
				e.setProperty(ProfileMetadata.ACTION, ProfileMetadata.ACTION_MAIL);
				e.setProperty(ProfileMetadata.VALUE, body);
				datastore.put(e);
				txn.commit();
			} finally {
				if (txn.isActive()) {
					txn.rollback();
				}
			}
			// return Mailer.sendMail(sub, body, email);
			return true;
		}
		return false;
	}

	// ------------- SUCCESS STORY BEGINS -------------------
	@SuppressWarnings("unused")
	@Override
	public void createSuccessStory(SuccessStory story) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Transaction txn = datastore.beginTransaction(withXG(true));
		try {
			validation: {
				List<Long> profileIds = findMyProfileIds(null);
				if (profileIds.contains(story.getBrideId()) || profileIds.contains(story.getGroomId())) {
					Query q = new Query(SuccessStory.class.getSimpleName());

					List<Query.Filter> l = new ArrayList<Query.Filter>();
					l.add(FilterOperator.EQUAL.of(SuccessStory.GROOM_ID, story.getGroomId()));
					l.add(FilterOperator.EQUAL.of(SuccessStory.BRIDE_ID, story.getBrideId()));
					addFiltersToQuery(q, l);
					q.setKeysOnly();
					List<Entity> results = datastore.prepare(q).asList(FetchOptions.Builder.withDefaults());
					Entity e = null;
					if (results.size() > 0) {
						for (Entity entity : results) {
							datastore.delete(entity.getKey());
						}
					}
					e = new Entity(SuccessStory.class.getSimpleName());
					setPropertiesToEntity(story, e);
					datastore.put(e);
				}
			}
			txn.commit();
		} finally {
			if (txn.isActive()) {
				txn.rollback();
			}
		}
	}

	@Override
	public List<Long> getAllSuccessStories() {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Query q = new Query(SuccessStory.class.getSimpleName());
		q.setFilter(FilterOperator.EQUAL.of(SuccessStory.APPROVED, Boolean.TRUE));
		q.setKeysOnly();

		List<Entity> results = datastore.prepare(q).asList(FetchOptions.Builder.withDefaults());
		List<Long> ids = new ArrayList<Long>();
		for (Entity e : results) {
			ids.add(e.getKey().getId());
		}
		return ids;
	}

	@Override
	public List<SuccessStory> getSuccessStories(List<Long> ids) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();

		List<Key> keys = createKeys(ids, SuccessStory.class.getSimpleName());

		Map<Key, Entity> entities = datastore.get(keys);
		List<SuccessStory> stories = new ArrayList<SuccessStory>();
		for (Map.Entry<Key, Entity> e : entities.entrySet()) {
			SuccessStory s = new SuccessStory();
			s.setId(e.getKey().getId());
			s.setProperties(e.getValue().getProperties());
			stories.add(s);
		}
		return stories;
	}

	@Override
	public Boolean approveSuccessStory(Long id) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Key k = createKeys(Arrays.asList(id), SuccessStory.class.getSimpleName()).get(0);
		Entity e;
		try {
			e = datastore.get(k);
			e.setProperty(SuccessStory.APPROVED, Boolean.TRUE);
			datastore.put(e);
			return Boolean.TRUE;
		} catch (EntityNotFoundException e1) {
			e1.printStackTrace();
		}
		return Boolean.FALSE;
	}

	@Override
	public Boolean deleteSuccessStory(Long id) {
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Transaction txn = datastore.beginTransaction(withXG(true));
		Boolean ret = Boolean.FALSE; 
		try {
			Key k = createKeys(Arrays.asList(id), SuccessStory.class.getSimpleName()).get(0);
			datastore.delete(k);
			txn.commit();
			ret = Boolean.TRUE;
		} finally {
			if (txn.isActive()) {
				txn.rollback();
			}
		}
		return ret;
	}

	// ------------- SUCCESS STORY ENDS -------------------

	public List<String> validateEntity(AbstractEntity<?> entity) {
		List<String> errors = new ArrayList<String>();
		if (entity instanceof Profile) {
			Profile p = (Profile) entity;
			if (p.getAboutMe().length() < 100) {
				errors.add("Please write something about yourself, at least 100 characters.");
			}

			if (p.getBloodGroup().isEmpty() || !ConstantsManager.getInstance().getList(Lists.BLOOD_GROUP).get("").contains(p.getBloodGroup())) {
				errors.add("Please provide blood group detail.");
			}

			if (p.getReligion().isEmpty() || !ConstantsManager.getInstance().getList(Lists.RELIGIONS).get("").contains(p.getReligion())) {
				errors.add("Please provide religion.");
			}
			if (p.getCaste().isEmpty() || !ConstantsManager.getInstance().getList(Lists.CASTE).get(p.getReligion()).contains(p.getCaste())) {
				errors.add("Please provide caste.");
			}

			if (p.getCurrCity().isEmpty() || p.getCurrState().isEmpty() || p.getCurrCountry().isEmpty() || p.getCurrLatitude() == null || p.getCurrLongitude() == null) {
				errors.add("Please provide your current address.");
			}

			if (p.getFamCity().isEmpty() || p.getFamState().isEmpty() || p.getFamCountry().isEmpty() || p.getFamLatitude() == null || p.getFamLongitude() == null) {
				errors.add("Please provide your family address.");
			}

			// TODO: check login
			if (p.getDegrees().isEmpty() || !ConstantsManager.getInstance().getList(Lists.DEGREES).get("").containsAll(p.getDegrees())) {
				errors.add("Please provide correct degrees.");
			}
			// TODO: check age
			// if(p.getDob() == null || ){
			// errors.add("Please provide correct degrees.");
			// }

			if (p.getEatingHabits().isEmpty() || !ConstantsManager.getInstance().getList(Lists.DIET).get("").contains(p.getEatingHabits())) {
				errors.add("Please provide diet information.");
			}

			if (p.getEmploymentType().isEmpty() || !ConstantsManager.getInstance().getList(Lists.EMPLOYMENT_TYPE).get("").contains(p.getEmploymentType())) {
				errors.add("Please provide employment type information.");
			}

			if (p.getGender().isEmpty() || !ConstantsManager.getInstance().getList(Lists.GENDER).get("").contains(p.getGender())) {
				errors.add("Please provide gender information.");
			}
			if (p.getHeight() == null) {
				errors.add("Please provide height information.");
			}
			// TODO: check it
			if (p.getLanguages().isEmpty() || !ConstantsManager.getInstance().getList(Lists.MOTHER_TONGUE).get("").containsAll(p.getLanguages())) {
				errors.add("Please provide known languages information.");
			}
			if (p.getManglik().isEmpty()) {
				errors.add("Please provide manglik information.");
			}
			if (p.getMaritalStatus().isEmpty() || !ConstantsManager.getInstance().getList(Lists.MARITAL_STATUS).get("").contains(p.getMaritalStatus())) {
				errors.add("Please provide known marital status information.");
			}
			if (p.getMaritalStatusChildren().isEmpty() || !ConstantsManager.getInstance().getList(Lists.MARITAL_STATUS_CHILDREN).get("").contains(p.getMaritalStatusChildren())) {
				errors.add("Please provide known marital status children information.");
			}
			if (p.getMotherTongue().isEmpty() || !ConstantsManager.getInstance().getList(Lists.MOTHER_TONGUE).get("").contains(p.getMotherTongue())) {
				errors.add("Please provide mothertongue information.");
			}
			if (p.getName().isEmpty()) {
				errors.add("Please provide name.");
			}
			if (p.getPhysicalStatus().isEmpty() || !ConstantsManager.getInstance().getList(Lists.PHY_STATUS).get("").contains(p.getPhysicalStatus())) {
				errors.add("Please provide physical status information.");
			}
			if (p.getPhysicalStatusDetail().isEmpty() || !ConstantsManager.getInstance().getList(Lists.PHY_STATUS_DETAIL).get("").contains(p.getPhysicalStatusDetail())) {
				errors.add("Please provide physical status detail information.");
			}
			if (p.getSmoking().isEmpty()) {
				errors.add("Please provide smoking information.");
			}
		}
		return errors;
	}

}
